package com.exchange.websocket.common.huobi;

import com.exchange.websocket.common.utils.redis.RedisUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import io.netty.handler.codec.http.websocketx.WebSocketVersion;
import io.netty.handler.ssl.SslContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.net.URI;
import java.util.*;


@SuppressWarnings("all")
public abstract class WebSocketBase {
    public static final Logger log = LoggerFactory.getLogger(WebSocketBase.class);
    private WebSocketService service = null;
    private Timer timerTask = null;
    private MoniterTask moniter = null;
    private EventLoopGroup group = null;
    private Bootstrap bootstrap = null;
    private Channel channel = null;
    private String url = null;
    private ChannelFuture future = null;
    private boolean isAlive = false;
    private String ping;
    private RedisUtil redisUtil;
    /**
     * 国内站siteFlag=0,国际站siteFlag=1
     */
    private int siteFlag = 0;
    private Set<String> subscribChannel = new HashSet<String>();

    public WebSocketBase(String url, WebSocketService serivce, RedisUtil redisUtil) {
        this.url = url;
        this.service = serivce;
        this.redisUtil =redisUtil;
    }

    public void start() {
        if (url == null) {
            log.info("WebSocketClient start error  url can not be null");
            return;
        }
        if (service == null) {
            log.info("WebSocketClient start error  WebSocketService can not be null");
            return;
        }
        moniter = new MoniterTask(this);
        this.connect();
        timerTask = new Timer();
        timerTask.schedule(moniter, 1000, 5000);
    }

    public void setStatus(boolean flag) {
        this.isAlive = flag;
    }

    public void addChannelOk(String channel) {
        if (channel == null) {
            return;
        }
        String dataMsg = "{'event':'addChannel','channel':'" + channel
                + "'}";
        this.sendMessage(dataMsg);
        subscribChannel.add(channel);
    }

    public void addChannelFCoin(String channel) {
        if (channel == null) {
            return;
        }
        String dataMsg = "{\"cmd\":\"sub\",\"args\":[\""+channel+"\"],\"id\":\"1\"}";
        log.info("dataMsg===>{}",dataMsg);
        this.sendMessage(dataMsg);
        subscribChannel.add(channel);
    }

    public void addChannelHitbic(String channel) {
        if (channel == null) {
            return;
        }
        String dataMsg = "{ \"method\": \"subscribeTicker\", \"params\": { \"symbol\": \""+channel+"\" }, \"id\": 1531299212498 }";
        log.info("dataMsg===>{}",dataMsg);
        this.sendMessage(dataMsg);
        subscribChannel.add(channel);
    }

    public void addChannelBittrex(String channel) {
        if (channel == null) {
            return;
        }
//        String dataMsg = "{\"H\": \"c2\", \"M\": \"SubscribeToExchangeDeltas\", \"A\": [\"BTC-ETH\"], \"I\": 0}";
        String dataMsg = "{\"SubscribeToExchangeDeltas\":[\"BTC-ETH\"]}";
        log.info("dataMsg===>{}",dataMsg);
        this.sendMessage(dataMsg);
        subscribChannel.add(channel);
    }

    public void addChannelHuobi(String channel) {
        if (channel == null) {
            return;
        }
        String dataMsg = "{\"sub\":\""+channel+"\",\"id\":\""+System.currentTimeMillis()+"\"}";
        log.info("dataMsg==>{}",dataMsg);
        this.sendMessage(dataMsg);
        subscribChannel.add(channel);
    }

    public void removeChannelOK(String channel) {
        if (channel == null) {
            return;
        }
        String dataMsg = "{'event':'removeChannel','channel':'" + channel
                + "'}";
        this.sendMessage(dataMsg);
        subscribChannel.remove(channel);
    }

    private void connect() {
        try {
            final URI uri = new URI(url);
            if (uri == null) {
                return;
            }
            if (uri.getHost().contains("com")) {
                siteFlag = 1;
            }
            group = new NioEventLoopGroup(1);
            bootstrap = new Bootstrap();
            final SslContext sslCtx = SslContext.newClientContext();
            final WebSocketClientHandler handler = new WebSocketClientHandler(
                    WebSocketClientHandshakerFactory.newHandshaker(uri,
                            WebSocketVersion.V13, null, false,
                            new DefaultHttpHeaders(), Integer.MAX_VALUE),
                    service, moniter,ping,redisUtil);
            bootstrap.group(group).option(ChannelOption.TCP_NODELAY, true)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline p = ch.pipeline();
                            if (sslCtx != null) {
                                p.addLast(sslCtx.newHandler(ch.alloc(),
                                        uri.getHost(), uri.getPort()));
                            }
                            p.addLast(new HttpClientCodec(),
                                    new HttpObjectAggregator(8192), handler);
                        }
                    });

            future = bootstrap.connect(uri.getHost(), uri.getPort());
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(final ChannelFuture future)
                        throws Exception {
                }
            });
            channel = future.sync().channel();
            handler.handshakeFuture().sync();
            this.setStatus(true);
        } catch (Exception e) {
            log.info("WebSocketClient start error ", e);
            group.shutdownGracefully();
            this.setStatus(false);
        }
    }

    private void sendMessage(String message) {
        if (!isAlive) {
            log.info("WebSocket is not Alive addChannel error");
        }
        channel.writeAndFlush(new TextWebSocketFrame(message));
    }

    public void sentPing() {

        String dataMsg = "{\"pong\":\""+redisUtil.get("huobiPong")+"\"}";
        log.info("pong==={}",redisUtil.get("huobiPong"));
        this.sendMessage(dataMsg);
    }

    public void reConnect() {
        try {
            this.group.shutdownGracefully();
            this.group = null;
            this.connect();
            if (future.isSuccess()) {
                this.setStatus(true);
                this.sentPing();
                Iterator<String> it = subscribChannel.iterator();
                while (it.hasNext()) {
                    String channel = it.next();
                    this.addChannelOk(channel);
                }

            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void setUrl(String url) {
        this.url = url;
    }

}
